package com.avicit.zipkinkafkaesserver.service;

import com.avicit.zipkinkafkaesserver.commands.InterfaceStatusDurationListCommand;
import com.avicit.zipkinkafkaesserver.utils.AtomicIntegerUtils;
import com.avicit.zipkinkafkaesserver.factory.DataListCommandFactory;
import com.avicit.zipkinkafkaesserver.commands.InterfaceDataListCommand;
import com.avicit.zipkinkafkaesserver.commands.InterfaceDurationListCommand;
import com.avicit.zipkinkafkaesserver.enums.ListTypeEnum;
import com.avicit.zipkinkafkaesserver.sparkapi.SparkDoCount;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import scala.Tuple2;
import zipkin2.Span;

import java.io.IOException;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;

@Service
public class SparkService {

    @Autowired
    private SpanService spanService;

//    @Autowired
//    private JavaSparkContext javaSparkContext;

    @Autowired
    private SparkConf sparkConf;


    /**
     * 请求接口统计 分时
     * @return
     * @throws IOException
     */
    public List<Tuple2<String, Integer>> interfaceCount(long startTime, long endTime, int limit) throws IOException {
        List<Span> spans = spanService.spanList(startTime,endTime, limit).get(ListTypeEnum.SERVER);
        return this.interfaceCount(spans);
    }

    /**
     * 请求接口统计 分时
     * @return
     * @throws IOException
     */
    public List<Tuple2<String, Integer>> interfaceCount(long startTime, long endTime, int limit, String interfaceName) throws IOException {

        List<Span> spans = StringUtils.isEmpty(interfaceName)
                ?spanService.spanList(startTime,endTime, limit).get(ListTypeEnum.SERVER)
                :spanService.spanList(startTime,endTime, limit,interfaceName).get(ListTypeEnum.SERVER);
        return this.interfaceCount(spans);
    }

    /**
     * 加入状态进行统计
     * @param startTime
     * @param endTime
     * @param limit
     * @param serverName
     * @param interfaceName
     * @return
     * @throws IOException
     */
    public List<Tuple2<String, Integer>> interfaceCountHasStatus(long startTime, long endTime, int limit, String serverName, String interfaceName) throws IOException {
        List<Span> spans = spanService.spanList(startTime,endTime, limit,serverName,interfaceName).get(ListTypeEnum.SERVER);
        return this.interfaceCountHasStatus(spans);
    }

    /**
     * 请求接口统计 分时
     * 返回map
     * @return
     * @throws IOException
     */
    public Map<String, Object> interfaceCountMap(long startTime, long endTime, int limit) throws IOException {
        List<Span> spans = spanService.spanList(startTime,endTime, limit).get(ListTypeEnum.SERVER);
        return this.interfaceCountMap(spans);
    }

    /**
     * 请求接口统计
     * @return
     * @throws IOException
     */
    public List<Tuple2<String, Integer>> interfaceCount() throws IOException {
        List<Span> spans = spanService.spanList().get(ListTypeEnum.SERVER);
        return this.interfaceCount(spans);
    }

    /**
     * 请求接口统计
     * @return
     * @throws IOException
     */
    public List<Tuple2<String, Integer>> interfaceCount(List<Span> spans) throws IOException {
        List<String> list = DataListCommandFactory.getInstance().create(InterfaceDataListCommand.class).execute(spans);
        //spark 计算返回结果
        return new SparkDoCount(sparkConf) {
            @Override
            protected List dataList() {
                return list.stream().map(item->{
                    String[] split = item.split("@");

                    return new StringBuffer()
                            .append(split[1]).append("@")//服务名
                            .append(split[2]).append("@")//请求方法
                            .append(split[3]).toString();//请求路径
                }).collect(Collectors.toList());
            }
        }.doCountResult();
    }
    /**
     * 请求接口统计 将状态统计进去
     * @return
     * @throws IOException
     */
    public List<Tuple2<String, Integer>> interfaceCountHasStatus(List<Span> spans) throws IOException {
        List<String> list = DataListCommandFactory.getInstance().create(InterfaceDataListCommand.class).execute(spans);
        //spark 计算返回结果
        return new SparkDoCount(sparkConf) {
            @Override
            protected List dataList() {
                return list.stream().map(item->{
                    String[] split = item.split("@");

                    return new StringBuffer()
                            .append(split[1]).append("@")//服务名
                            .append(split[2]).append("@")//请求方法
                            .append(split[3]).append("@")//请求路径
                            .append(split[4])//接口状态
                            .toString();
                }).collect(Collectors.toList());
            }
        }.doCountResult();
    }

    /**
     * 请求接口统计
     * @return
     * @throws IOException
     */
    public Map<String,Object> interfaceCountMap(List<Span> spans) throws IOException {
        Map<String,Object> resultMap = new HashMap<>();
        List<String> list = DataListCommandFactory.getInstance().create(InterfaceDataListCommand.class).execute(spans);
        //spark 计算返回结果
        List<String> collect = list.stream().map(item -> {
            String[] split = item.split("@");
            return new StringBuffer()
                    .append(split[0]).append("@")
                    .append(split[1]).append("@")
                    .append(split[2]).append("@")
                    .append(split[3]).append("@")
                    .append(split[4]).toString();
        }).collect(Collectors.toList());//计算时不需要时间戳
        List<Tuple2<String, Integer>> countResult = new SparkDoCount(sparkConf) {
            @Override
            protected List dataList() {
                return collect;
            }
        }.doCountResult();
        if (countResult.size() == 0) {//没有统计结果，直接返回
            resultMap.put("countResult",countResult);
            return resultMap;
        }
        //取时间
        Long endTimeFromEsRecord = list.stream().map(item -> Long.valueOf(item.split("@")[5])).collect(Collectors.toList()).stream().sorted(Comparator.comparing(Long::intValue).reversed()).findFirst().get();
        resultMap.put("countResult",countResult);
        resultMap.put("lastEndTime",endTimeFromEsRecord);
        return resultMap;
    }


    /**
     * 接口耗时平均统计
     * @return
     */
    public Map<String, Object> interfaceDurationCountAvg(long startTime, long endTime, int limit) throws IOException {
        List<Span> spans = spanService.spanList(startTime, endTime,limit).get(ListTypeEnum.SERVER);
       return interfaceDurationCount(spans);

    }

    /**
     * 接口耗时平均统计
     * @return
     */
    public Map<String, Object> interfaceDurationCount() throws IOException {
        List<Span> spans = spanService.spanList().get(ListTypeEnum.SERVER);
        return this.interfaceDurationCount(spans);
    }

    /**
     * 接口耗时平均统计
     * @return
     */
    public Map<String, Object> interfaceDurationCount(long startTime, long endTime, int limit, String interfaceName) throws IOException {
        List<Span> spans = spanService.spanList(startTime,endTime, limit ,interfaceName).get(ListTypeEnum.SERVER);
        return this.interfaceDurationCount(spans);
    }

    /**
     * 接口耗时平均统计
     * @return
     */
    public Map<String, Object> interfaceDurationCountNoIp(long startTime, long endTime, int limit, String interfaceName) throws IOException {
        List<Span> spans = spanService.spanList(startTime,endTime, limit ,interfaceName).get(ListTypeEnum.SERVER);
        return this.interfaceDurationCountNoIp(spans);
    }
    /**
     * 接口耗时平均统计
     * @return
     */
    public Map<String, Object> interfaceDurationCountNoIp(long startTime, long endTime, int limit, String serverName,String interfaceName) throws IOException {
        List<Span> spans = spanService.spanList(startTime,endTime, limit ,serverName, interfaceName).get(ListTypeEnum.SERVER);
        return this.interfaceDurationCountNoIp(spans);
    }

    /**
     * 接口耗时平均统计
     * @return
     */
    public Map<String, Object> interfaceDurationCount(List<Span> spans) throws IOException {
        Map<String, Object> resultMap = new HashMap<>();
        List<Tuple2<String,Double>> execute = DataListCommandFactory.getInstance().create(InterfaceDurationListCommand.class).execute(spans);
        List<Tuple2<String, Double>> averageList = new SparkDoCount(sparkConf) {
            @Override
            protected List dataList() {
                return execute.stream().map(item->{
                    String[] split = item._1.split("@");
                    return new Tuple2<String, Double>(new StringBuffer()
                            .append(split[0]).append("@")
                            .append(split[1]).append("@")
                            .append(split[2]).append("@")
                            .append(split[3]).toString(), item._2);
                }).collect(Collectors.toList());//统计的时候 去掉时间搓
            }
        }.averageForDataList();
        if (averageList.size() == 0) {//没有统计结果，直接返回
            resultMap.put("countResult",averageList);
            return resultMap;
        }
        //取时间
        Long endTimeFromEsRecord = execute.stream().map(item -> Long.valueOf(item._1.split("@")[4])).collect(Collectors.toList()).stream().sorted(Comparator.comparing(Long::intValue).reversed()).findFirst().get();
        resultMap.put("countResult",averageList);
        resultMap.put("lastEndTime",endTimeFromEsRecord);
         return resultMap;
    }
    /**
     * 接口耗时平均统计
     * @return
     */
    public Map<String, Object> interfaceDurationCountNoIp(List<Span> spans) throws IOException {
        Map<String, Object> resultMap = new HashMap<>();
        List<Tuple2<String,Double>> execute = DataListCommandFactory.getInstance().create(InterfaceDurationListCommand.class).execute(spans);
        List<Tuple2<String, Double>> averageList = new SparkDoCount(sparkConf) {
            @Override
            protected List dataList() {
                return execute.stream().map(item->{
                    String[] split = item._1.split("@");
                    return new Tuple2<String, Double>(new StringBuffer()
//                            .append(split[0]).append("@")
                            .append(split[1]).append("@")
                            .append(split[2]).append("@")
                            .append(split[3]).toString(), item._2);
                }).collect(Collectors.toList());//统计的时候 去掉时间搓
            }
        }.averageForDataList();
        if (averageList.size() == 0) {//没有统计结果，直接返回
            resultMap.put("countResult",averageList);
            return resultMap;
        }
        //取时间
        Long endTimeFromEsRecord = execute.stream().map(item -> Long.valueOf(item._1.split("@")[4])).collect(Collectors.toList()).stream().sorted(Comparator.comparing(Long::intValue).reversed()).findFirst().get();
        resultMap.put("countResult",averageList);
        resultMap.put("lastEndTime",endTimeFromEsRecord);
         return resultMap;
    }

    /**
     * 耗时列表 没有状态
     * max：最大耗时
     * min： 最小耗时
     * 这个可以优化
     * @return
     */
    public List<Tuple2<String,Double>> interfaceDurationCountList(double min, double max,int topN, List<Span> spans) throws IOException {
        List<Tuple2<String,Double>> execute = DataListCommandFactory.getInstance().create(InterfaceDurationListCommand.class).execute(spans);
        return getTuple2s(min, max, topN, execute);
    }

    /**
     * 耗时列表 有状态
     * max：最大耗时
     * min： 最小耗时
     * 这个可以优化
     * @return
     */
    public List<Tuple2<String,Double>> interfaceDurationCountListStatus(double min, double max,int topN, List<Span> spans) throws IOException {
        List<Tuple2<String,Double>> execute = DataListCommandFactory.getInstance().create(InterfaceStatusDurationListCommand.class).execute(spans);
        return getTuple2s(min, max, topN, execute);
    }

    /**
     * 执行耗时列表统计
     * @param min
     * @param max
     * @param topN
     * @param execute
     * @return
     */
    private List<Tuple2<String, Double>> getTuple2s(double min, double max, int topN, List<Tuple2<String, Double>> execute) {
        List<Tuple2<String, Double>> tuple2List = execute.stream().filter(stringDoubleTuple2 -> {
            if (max == 0 && min != 0) {
                return stringDoubleTuple2._2 >= min;
            }
            if (min == 0 && max != 0) {
                return stringDoubleTuple2._2 <= max;
            }
            if (min != 0 && max != 0) {
                return stringDoubleTuple2._2 >= min && stringDoubleTuple2._2 <= max;
            }
            return stringDoubleTuple2._2 != 0;
        }).collect(Collectors.toList());
        if (topN>0) {//topN 排序
            Comparator<Tuple2<String,Double>> comparator = Comparator.comparing(tuple2 -> tuple2._2);
            List<Tuple2<String,Double>> collect = tuple2List.stream().sorted(comparator.reversed()).collect(Collectors.toList());

            //设置排名
            List<Tuple2<Tuple2<String, Double>, Integer>> tuple2s = collect.stream().map(stringDoubleTuple2 -> {

                return new Tuple2<Tuple2<String, Double>, Integer>(stringDoubleTuple2, AtomicIntegerUtils.add());
            }).collect(Collectors.toList());

            topN=topN>=collect.size()?collect.size():topN;

            List<Tuple2<String, Double>> topList = tuple2s.subList(0,topN).stream().map(item-> new Tuple2<String,Double>(item._1._1,item._1._2)).collect(Collectors.toList());
            return topList;
        }
        return tuple2List;
    }

    /**
     * 耗时列表
     * max：最大耗时
     * min： 最小耗时
     * 这个可以优化
     * @return
     */
    public List<Tuple2<String,Double>> interfaceDurationCountList(double min, double max,int topN) throws IOException {
        List<Span> spans = spanService.spanList().get(ListTypeEnum.SERVER);
        return this.interfaceDurationCountList(min, max, topN, spans);
    }

    /**
     * 耗时列表
     * max：最大耗时
     * min： 最小耗时
     * 这个可以优化
     * @return
     */
    public List<Tuple2<String,Double>> interfaceDurationCountList(double min, double max,int topN,long startTime, long endTime, String interfaceName) throws IOException {
        List<Span> spans = spanService.spanList(startTime,endTime,Integer.MAX_VALUE, interfaceName).get(ListTypeEnum.SERVER);
        return this.interfaceDurationCountList(min, max, topN, spans);
    }

    /**
     * 耗时列表
     * max：最大耗时
     * min： 最小耗时
     * 这个可以优化
     * @return
     */
    public List<Tuple2<String,Double>> interfaceDurationCountList(long startTime, long endTime,int limit) throws IOException {
        List<Span> spans = spanService.spanList(startTime, endTime, limit).get(ListTypeEnum.SERVER);
        //定时任务不对结果进行过滤筛选
        return this.interfaceDurationCountList(0, 0, 0, spans);
    }
    /**
     * 耗时列表
     * max：最大耗时
     * min： 最小耗时
     * 这个可以优化
     * @return
     */
    public List<Tuple2<String,Double>> interfaceDurationCountList(long startTime, long endTime,int limit,String serverName, String interfaceName) throws IOException {
        List<Span> spans = spanService.spanList(startTime, endTime, limit,serverName,interfaceName).get(ListTypeEnum.SERVER);
        //不对结果进行过滤筛选
        return this.interfaceDurationCountListStatus(0, 0, 0, spans);
    }


}
